package com.geekbg.zookeeper.distributedqueue;

import com.alibaba.fastjson.JSONObject;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.queue.BlockingQueueConsumer;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.utils.CloseableUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author pizhihui
 * @date 2023-02-27 20:11
 */
public class Client2 {
    private static final QueueSerializer<TestQueueItem> serializer = new QueueItemSerializer();

    private static final String QUEUE_PATH = "/a/queue";


    public static void main(String[] args) {
        final int itemQty = 10;

        String connString = "10.9.21.39:2181";
        CuratorFramework client = CuratorFrameworkFactory.newClient(connString, new RetryOneTime(1));

        client.start();


        DistributedQueue<TestQueueItem> queue = null;
        try {
            BlockingQueueConsumer<TestQueueItem> consumer = new BlockingQueueConsumer<TestQueueItem>(new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState) {
                    System.out.println("改变了......." + newState);
                }
            });

            queue = QueueBuilder.builder(client, consumer, serializer, QUEUE_PATH)
                    .lockPath("/a/locks").
                    buildQueue();
            queue.start();

            // 模拟producer
            QueueTestProducer producer = new QueueTestProducer(queue, itemQty, 0);
            ExecutorService service = Executors.newCachedThreadPool();
            service.submit(producer);

            int iteration = 0;

            while (consumer.size() < itemQty) {
//                if (++iteration > itemQty) {
//                    break;
//                }
                System.out.println("等待放入...." + consumer.size());
                Thread.sleep(1000);
            }

            int size = consumer.size();

            List<TestQueueItem> items = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                TestQueueItem take = consumer.take();
                items.add(take);
            }
//            List<QueueItem> items = consumer.getItems();

            System.out.println("获得大小是: "+ items.size());

            System.out.println(JSONObject.toJSONString(items));

        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
        }


    }

}
